from typing import Optional, List, Dict


def format_udf_udaf(args: List, udfs: [List[Dict[str, str]]], udafs: [List[Dict[str, str]]]):
    if udfs is not None:
        i = 1
        for udf in udfs:
            name = udf['name']
            udf_class = udf['udfClass']
            args.append("-reader.udf." + str(i) + ".name=" + name)
            args.append("-reader.udf." + str(i) + ".udfClass=" + udf_class)
            i += 1
    if udafs is not None:
        i = 1
        for udaf in udafs:
            name = udaf['name']
            udaf_class = udaf['udafClass']
            args.append("-reader.udaf." + str(i) + ".name=" + name)
            args.append("-reader.udaf." + str(i) + ".udafClass=" + udaf_class)
            i += 1


class HiveArgsModel(object):
    def __init__(
            self,
            streaming: Optional[bool] = None,
            reader_type: Optional[str] = None,
            reader_sql: Optional[str] = None,
            reader_db_tables: Optional[str] = None,
            reader_udfs: Optional[List[Dict[str, str]]] = None,
            reader_udafs: Optional[List[Dict[str, str]]] = None,
            writer_type: Optional[str] = None,
            writer_path: Optional[str] = None,
            writer_partition_by: Optional[str] = None,
            writer_repartitions: Optional[int] = None,
            writer_db: Optional[str] = None,
            writer_table: Optional[str] = None,
            writer_format: Optional[str] = None,
            writer_clean_table: Optional[bool] = None,
    ) -> None:
        self._streaming = streaming
        self._reader_type = reader_type
        self._reader_sql = reader_sql
        self._reader_db_tables = reader_db_tables
        self._reader_udfs = reader_udfs
        self._reader_udafs = reader_udafs
        self._writer_type = writer_type
        self._writer_path = writer_path
        self._writer_partition_by = writer_partition_by
        self._writer_repartitions = writer_repartitions
        self._writer_db = writer_db
        self._writer_table = writer_table
        self._writer_format = writer_format
        self._writer_clean_table = writer_clean_table

    def to_list_reader(self):
        args = [
            "-streaming=" + str(self._streaming),
            "-reader.type=" + self._reader_type,
            "-reader.sql=" + self._reader_sql,
            "-reader.databaseAndTables=" + self._reader_db_tables
        ]
        format_udf_udaf(args, self._reader_udfs, self._reader_udafs)
        return args

    def to_list_writer(self):
        list_writer = [
            "-writer.type=" + self._writer_type,
            "-writer.path=" + self._writer_path,
            "-writer.format=" + self._writer_format
        ]
        if self._writer_partition_by:
            list_writer.append("-writer.partitionBy=" + self._writer_partition_by)

        if self._writer_repartitions and self._writer_repartitions != '':
            list_writer.append("-writer.repartitions=" + str(self._writer_repartitions))
        return list_writer

    @property
    def reader_udafs(self):
        return self._reader_udafs

    @property
    def reader_udfs(self):
        return self._reader_udfs


class JdbcArgsModel(object):
    def __init__(
            self,
            streaming: Optional[bool] = None,
            reader_type: Optional[str] = None,
            reader_sql: Optional[str] = None,
            reader_db_tables: Optional[str] = None,
            reader_ip: Optional[str] = None,
            reader_port: Optional[int] = None,
            reader_user: Optional[str] = None,
            reader_password: Optional[str] = None,
            reader_udfs: Optional[List[Dict[str, str]]] = None,
            reader_udafs: Optional[List[Dict[str, str]]] = None,
            writer_type: Optional[str] = None,
            writer_db: Optional[str] = None,
            writer_table: Optional[str] = None,
            writer_clean_table: Optional[bool] = None,
            writer_ip: Optional[str] = None,
            writer_port: Optional[int] = None,
            writer_user: Optional[str] = None,
            writer_password: Optional[str] = None,
    ) -> None:
        self._streaming = streaming
        self._reader_type = reader_type
        self._reader_sql = reader_sql
        self._reader_db_tables = reader_db_tables
        self._reader_ip = reader_ip
        self._reader_port = reader_port
        self._reader_user = reader_user
        self._reader_password = reader_password
        self._reader_udfs = reader_udfs
        self._reader_udafs = reader_udafs
        self._writer_type = writer_type
        self._writer_db = writer_db
        self._writer_table = writer_table
        self._writer_clean_table = writer_clean_table
        self._writer_ip = writer_ip
        self._writer_port = writer_port
        self._writer_user = writer_user
        self._writer_password = writer_password

    def to_list_reader(self):
        args = [
            "-streaming=" + str(self._streaming),
            "-reader.type=" + self._reader_type,
            "-reader.sql=" + self._reader_sql,
            "-reader.ip=" + self._reader_ip,
            "-reader.port=" + str(self._reader_port),
            "-reader.user=" + self._reader_user,
            "-reader.password=" + self._reader_password,
            "-reader.databaseAndTables=" + self._reader_db_tables
        ]
        format_udf_udaf(args, self._reader_udfs, self._reader_udafs)
        return args

    def to_list_writer(self):
        list_writer = [
            "-writer.type=" + self._writer_type,
            "-writer.user=" + self._writer_user,
            "-writer.password=" + self._writer_password,
            "-writer.port=" + str(self._writer_port),
            "-writer.ip=" + self._writer_ip,
            "-writer.table=" + self._writer_table,
            "-writer.database=" + self._writer_db,
            "-writer.cleanTable=" + str(self._writer_clean_table),
        ]
        return list_writer

    @property
    def reader_udfs(self):
        return self._reader_udfs

    @property
    def reader_udafs(self):
        return self._reader_udafs


class KafkaArgsModel(object):
    def __init__(
            self,
            streaming: Optional[bool] = None,
            reader_type: Optional[str] = None,
            reader_servers: Optional[str] = None,
            reader_topics: Optional[str] = None,
            reader_field_types: Optional[str] = None,
            reader_max_bytes: Optional[int] = None,
            reader_max_offset_pertrigger: Optional[int] = None,
            reader_decoder_class: Optional[str] = None,

            writer_type: Optional[str] = None,
            writer_servers: Optional[str] = None,
            writer_topic: Optional[str] = None,
            writer_acks: Optional[str] = None,
    ) -> None:
        self._streaming = streaming
        self._reader_type = reader_type
        self._reader_servers = reader_servers
        self._reader_topics = reader_topics
        self._reader_field_types = reader_field_types
        self._reader_max_bytes = reader_max_bytes
        self._reader_max_offset_pertrigger = reader_max_offset_pertrigger
        self._reader_decoder_class = reader_decoder_class

        self._writer_servers = writer_servers
        self._writer_topic = writer_topic
        self._writer_type = writer_type
        self._writer_acks = writer_acks

    def to_list_reader(self):
        args = [
            "-streaming=" + str(self._streaming),
            "-reader.type=" + self._reader_type,
            "-reader.kafka.bootstrap.servers=" + self._reader_servers,
            "-reader.topics=" + self._reader_topics,
            "-reader.fieldTypes=" + self._reader_field_types,
            "-reader.decoderClass=" + self._reader_decoder_class,
        ]
        if self._reader_max_bytes and self._reader_max_bytes > 0:
            args.append("-reader.kafka.max.partition.fetch.bytes=" + str(self._reader_max_bytes))
        if self._reader_max_offset_pertrigger and self._reader_max_offset_pertrigger > 0:
            args.append("-reader.kafka.maxOffsetsPerTrigger=" + str(self._reader_max_offset_pertrigger))
        return args

    def to_list_writer(self):
        list_writer = [
            "-writer.type=" + self._writer_type,
            "-writer.kafka.bootstrap.servers=" + self._writer_servers,
            "-writer.topic=" + self._writer_topic
            ]
        if self._writer_acks:
            list_writer.append("-writer.kafka.acks=" + self._writer_acks)
        return list_writer


class DorisModel(object):
    def __init__(
            self,
            streaming: Optional[bool] = None,
            writer_type: Optional[str] = None,
            writer_fenodes: Optional[str] = None,
            writer_table: Optional[str] = None,
            writer_user: Optional[str] = None,
            writer_password: Optional[str] = None,
    ) -> None:
        self._streaming = streaming
        self._writer_type = writer_type
        self._writer_fenodes = writer_fenodes
        self._writer_table = writer_table
        self._writer_user = writer_user
        self._writer_password = writer_password

    def to_list_writer(self):
        list_writer = [
            "-writer.type=" + self._writer_type,
            "-writer.fenodes=" + self._writer_fenodes,
            "-writer.table=" + self._writer_table,
            "-writer.user=" + self._writer_user,
            "-writer.password=" + self._writer_password
        ]
        return list_writer


class EsModel(object):
    def __init__(
            self,
            streaming: Optional[bool] = None,
            nodes: Optional[str] = None,
            port: Optional[int] = None,
            user: Optional[str] = None,
            password: Optional[str] = None,
            writer_type: Optional[str] = None,
            writer_index: Optional[str] = None,
            writer_index_type: Optional[str] = None,
    ) -> None:
        self._streaming = streaming
        self._writer_type = writer_type
        self._writer_index = writer_index
        self._writer_index_type = writer_index_type
        self._nodes = nodes
        self._port = port
        self._user = user
        self._password = password

    def to_list_writer(self):
        list_writer = [
            "-writer.indexType=" + self._writer_index_type,
            "-writer.index=" + self._writer_index,
            "-writer.nodes=" + self._nodes,
            "-writer.port=" + str(self._port),
            "-writer.authPassword=" + self._password,
            "-writer.authUser=" + self._user,
            "-writer.type=" + self._writer_type,
            "-writer.isJsonWriter=" + str(True)
        ]
        return list_writer